#pragma once

#include <brpc/server.h>
#include <butil/logging.h>
#include <cctype>
#include <regex>

#include "etcd.hpp"
#include "logger.hpp"
#include "base.pb.h"
#include "user.pb.h"
#include "file.pb.h"
#include "utils.hpp"
#include "message.pb.h"
#include "es_data.hpp"
#include "channel.hpp"
#include "mysql_message.hpp"
#include "rabbitmq.hpp"


 
namespace xu
{
   
    class MsgStorageServicelmpl :public xu::MsgStorageService
    {
    public:
        MsgStorageServicelmpl(
        ServiceManage::ptr service_manage,
        std::shared_ptr<elasticlient::Client> es_client,
        std::shared_ptr<odb::mysql::database> mysql_client,
        MQClient::ptr   mq_client,
        const std::string &user_service_name,
        const std::string &file_service_name) 
        :_service_manage(service_manage),
        _es_message(std::make_shared<ESMessage>(es_client)),
        _mysql_message(std::make_shared<ManageMessage>(mysql_client)),
        _mq_client(mq_client),
        _user_service_name(user_service_name),
        _file_service_name(file_service_name)
        {}
        ~MsgStorageServicelmpl(){}
        
      
        virtual void GetHistoryMsg(::google::protobuf::RpcController* controller,
                       const ::xu::GetHistoryMsgReq* request,
                       ::xu::GetHistoryMsgRsp* response,
                       ::google::protobuf::Closure* done)  override
        {
            brpc::ClosureGuard clo(done);
            response->set_request_id(request->request_id()); 
            auto err_rsp = [this,response](const std::string & err_msg)
            {
                response->set_success(false);
                response->set_errmsg(err_msg);
            };

            //1. 从请求中，获取会话 ID， 要获取的消息的起始时间与结束时间。
            std::string ssid = request->chat_session_id();
            boost::posix_time::ptime stime = boost::posix_time::from_time_t(request->start_time());
            boost::posix_time::ptime etime = boost::posix_time::from_time_t(request->over_time());
            //2. 访问数据库，从数据库中按时间进行范围查询，获取消息简略信息（消息 ID，会
            //话 ID，消息类型，产生时间，发送者用户 ID，文本消息内容，文件消息元信息）
            auto res = _mysql_message->SelectMessageRange(ssid,stime,etime);
            if(res.empty())
            {
                response->set_success(true);
                return;
            }

            //3. 循环构造完整消息（从用户子服务获取消息的发送者用户信息，从文件子服务获
            //取文件/语音/图片数据）
            //从用户子服务获取消息的发送者用户信息
            auto user_channel = _service_manage->Choose(_user_service_name);
            if(!user_channel)
            {
                LOG_ERROR("获取用户服务信道失败");
                err_rsp("获取用户服务信道失败");
                return;
            }

            xu::UserService_Stub user_stub(user_channel.get());
            xu::GetMultiUserInfoReq user_req;
            xu::GetMultiUserInfoRsp user_rsp;
            brpc::Controller user_cntl;

            std::string uid = xu::uuid();
            user_req.set_request_id(uid);
            std::unordered_set<std::string> user_ids;
            for( auto& msg : res)
            {
                user_ids.insert(msg.from_user_id());
            }

            for(auto & s : user_ids)
            {
                user_req.add_users_id(s);
            }
            user_stub.GetMultiUserInfo(& user_cntl,&user_req,&user_rsp,nullptr);
            if( user_cntl.Failed() || user_rsp.success() == false)
            {
                LOG_ERROR("获取用户信息失败");
                err_rsp("获取用户信息失败");
                return;
            }


            //从文件子服务获取文件/语音/图片数据
            auto file_channel = _service_manage->Choose(_file_service_name);
            if (!file_channel) 
            {
                LOG_ERROR("获取文件服务信道失败");
                err_rsp("获取文件服务信道失败");
                return;
            }

            xu::FileService_Stub file_stub(file_channel.get());
            xu::GetMultiFileReq file_req;
            xu::GetMultiFileRsp file_rsp;
            brpc::Controller file_cntl;
            file_req.set_request_id(uuid());

            for( auto& msg : res)
            {
                if(!msg.file_id().empty())
                {
                    file_req.add_file_id_list(msg.file_id());
                }
            }
            file_stub.GetMultiFile(&file_cntl,&file_req,&file_rsp,nullptr);
            if(file_cntl.Failed() || file_rsp.success() == false)
            {
                LOG_ERROR("获取文件信息失败");
                err_rsp("获取文件信息失败");
                return;
            }
            // 4. 组织响应返回给网关服务器
            auto user_map = user_rsp.users_info();
            auto file_map = file_rsp.file_data();
            response->set_success(true);
            for( auto& msg : res)
            {
                auto message_info = response->add_msg_list();
                message_info->set_message_id(msg.message_id());
                message_info->set_chat_session_id(msg.to_session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(msg.created_time()));
                message_info->mutable_sender()->CopyFrom(user_map[msg.from_user_id()]);

                switch(msg.message_type())
                {
                    case MessageType::STRING :
                    {
                        message_info->mutable_message()->set_message_type(MessageType::STRING);
                        message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
                    }
                    break;
                    case MessageType::IMAGE :
                    {
                        message_info->mutable_message()->set_message_type(MessageType::IMAGE);
                        message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());
                        message_info->mutable_message()->mutable_image_message()->set_image_content(file_map[msg.file_id()].file_content());
                    }
                    break;
                    case MessageType::FILE :
                    {
                        message_info->mutable_message()->set_message_type(MessageType::FILE);
                        message_info->mutable_message()->mutable_file_message()->set_file_name(msg.filename());
                        message_info->mutable_message()->mutable_file_message()->set_file_size(msg.filesize());
                        message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());
                        message_info->mutable_message()->mutable_file_message()->set_file_contents(file_map[msg.file_id()].file_content());
                    }
                    break;
                    case MessageType::SPEECH :
                    {
                        message_info->mutable_message()->set_message_type(MessageType::SPEECH);
                        message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());
                        message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_map[msg.file_id()].file_content());
                    }
                    break;
                    default:
                    LOG_ERROR("消息类型错误");
                    return;
                }
            }
        }
        virtual void GetRecentMsg(::google::protobuf::RpcController* controller,
                       const ::xu::GetRecentMsgReq* request,
                       ::xu::GetRecentMsgRsp* response,
                       ::google::protobuf::Closure* done)  override
        {
            brpc::ClosureGuard clo(done);
            response->set_request_id(request->request_id()); 
            auto err_rsp = [this,response](const std::string & err_msg)
            {
                response->set_success(false);
                response->set_errmsg(err_msg);
            };

            // 1. 从请求中，获取会话 ID， 要获取的消息数量
            std::string ssid = request->chat_session_id();
            int count = request->msg_count();

            // 2. 访问数据库，从数据库中按时间排序，获取指定数量的消息简略信息（消息 ID，
            // 会话 ID，消息类型，产生时间，发送者用户 ID，文本消息内容，文件消息元信息）
            auto res = _mysql_message->SelectMessageN(ssid,count);
            if(res.empty())
            {
                response->set_success(true);
                return;
            }

            // 3. 循环构造完整消息
            //从用户子服务获取消息的发送者用户信息
            auto user_channel = _service_manage->Choose(_user_service_name);
            if(!user_channel)
            {
                LOG_ERROR("获取用户服务信道失败");
                err_rsp("获取用户服务信道失败");
                return;
            }

            xu::UserService_Stub user_stub(user_channel.get());
            xu::GetMultiUserInfoReq user_req;
            xu::GetMultiUserInfoRsp user_rsp;
            brpc::Controller user_cntl;

            std::string uid = xu::uuid();
            user_req.set_request_id(uid);
            std::unordered_set<std::string> user_ids;
            for( auto& msg : res)
            {
                user_ids.insert(msg.from_user_id());
            }

            for(auto & s : user_ids)
            {
                user_req.add_users_id(s);
            }
            user_stub.GetMultiUserInfo(& user_cntl,&user_req,&user_rsp,nullptr);
            if( user_cntl.Failed() || user_rsp.success() == false)
            {
                LOG_ERROR("获取用户信息失败");
                err_rsp("获取用户信息失败");
                return;
            }


            //从文件子服务获取文件/语音/图片数据
            auto file_channel = _service_manage->Choose(_file_service_name);
            if (!file_channel) 
            {
                LOG_ERROR("获取文件服务信道失败");
                err_rsp("获取文件服务信道失败");
                return;
            }

            xu::FileService_Stub file_stub(file_channel.get());
            xu::GetMultiFileReq file_req;
            xu::GetMultiFileRsp file_rsp;
            brpc::Controller file_cntl;
            file_req.set_request_id(uuid());

            for( auto& msg : res)
            {
                if(!msg.file_id().empty())
                {
                    file_req.add_file_id_list(msg.file_id());
                }
            }
            file_stub.GetMultiFile(&file_cntl,&file_req,&file_rsp,nullptr);
            if(file_cntl.Failed() || file_rsp.success() == false)
            {
                LOG_ERROR("获取文件信息失败");
                err_rsp("获取文件信息失败");
                return;
            }
            // 4. 组织响应返回给网关服务器
            auto user_map = user_rsp.users_info();
            auto file_map = file_rsp.file_data();
            response->set_success(true);
            for(auto& msg : res)
            {
                auto message_info = response->add_msg_list();
                message_info->set_message_id(msg.message_id());
                message_info->set_chat_session_id(msg.to_session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(msg.created_time()));
                message_info->mutable_sender()->CopyFrom(user_map[msg.from_user_id()]);

                switch(msg.message_type())
                {
                    case MessageType::STRING :
                    {
                        message_info->mutable_message()->set_message_type(MessageType::STRING);
                        message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
                    }
                    break;
                    case MessageType::IMAGE :
                    {
                        message_info->mutable_message()->set_message_type(MessageType::IMAGE);
                        message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());
                        message_info->mutable_message()->mutable_image_message()->set_image_content(file_map[msg.file_id()].file_content());
                    }
                    break;
                    case MessageType::FILE :
                    {
                        message_info->mutable_message()->set_message_type(MessageType::FILE);
                        message_info->mutable_message()->mutable_file_message()->set_file_name(msg.filename());
                        message_info->mutable_message()->mutable_file_message()->set_file_size(msg.filesize());
                        message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());
                        message_info->mutable_message()->mutable_file_message()->set_file_contents(file_map[msg.file_id()].file_content());
                    }
                    break;
                    case MessageType::SPEECH :
                    {
                        message_info->mutable_message()->set_message_type(MessageType::SPEECH);
                        message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());
                        message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_map[msg.file_id()].file_content());
                    }
                    break;
                    default:
                    LOG_ERROR("消息类型错误");
                    return;
                }
            }
        }
        virtual void MsgSearch(::google::protobuf::RpcController* controller,
                       const ::xu::MsgSearchReq* request,
                       ::xu::MsgSearchRsp* response,
                       ::google::protobuf::Closure* done)  override
        {
            brpc::ClosureGuard clo(done);
            response->set_request_id(request->request_id()); 
            auto err_rsp = [this,response](const std::string & err_msg)
            {
                response->set_success(false);
                response->set_errmsg(err_msg);
            };
            // 1. 从请求中，获取会话 ID， 搜索关键字
            std::string ssid = request->chat_session_id();
            std::string skey = request->search_key();
            // 2. 基于封装的 ES 客户端，访问 ES 服务器进行文本消息搜索（以消息内容进行搜索，
            // 以会话 ID 进行过滤），从 ES 服务器获取到消息简息（消息 ID，会话 ID， 文本消息内
            // 容）。
            auto res = _es_message->Search(skey,ssid);
            if(res.empty())
            {
                response->set_success(true);
                return;
            }
            // 3. 循环从数据库根据消息 ID 获取消息简息（消息 ID，消息类型，会话 ID，发送者
            // ID，产生时间，文本消息内容，文件消息元数据）。
            auto user_channel = _service_manage->Choose(_user_service_name);
            if(!user_channel)
            {
                LOG_ERROR("获取用户服务信道失败");
                err_rsp("获取用户服务信道失败");
                return;
            }

            xu::UserService_Stub user_stub(user_channel.get());
            xu::GetMultiUserInfoReq user_req;
            xu::GetMultiUserInfoRsp user_rsp;
            brpc::Controller user_cntl;

            std::string uid = xu::uuid();
            user_req.set_request_id(uid);
            std::unordered_set<std::string> user_ids;
            for( auto& msg : res)
            {
                user_ids.insert(msg.from_user_id());
            }

            for(auto & s : user_ids)
            {
                user_req.add_users_id(s);
            }
            user_stub.GetMultiUserInfo(& user_cntl,&user_req,&user_rsp,nullptr);
            if( user_cntl.Failed() || user_rsp.success() == false)
            {
                LOG_ERROR("获取用户信息失败");
                err_rsp("获取用户信息失败");
                return;
            }

            // 5. 组织响应返回给网关服务器。
            auto user_map = user_rsp.users_info();
            response->set_success(true);
            for( auto& msg : res)
            {
                auto message_info = response->add_msg_list();
                message_info->set_message_id(msg.message_id());
                message_info->set_chat_session_id(msg.to_session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(msg.created_time()));
                message_info->mutable_sender()->CopyFrom(user_map[msg.from_user_id()]);
                message_info->mutable_message()->set_message_type(MessageType::STRING);
                message_info->mutable_message()->mutable_string_message()->set_content(msg.content());

            }
        }


        void onMessage(const char * body, uint64_t size)
        {
            LOG_DEBUG("收到新消息，进行存储处理！");
            //1.将消息反序列化
            xu::MessageInfo message;
            bool ret = message.ParseFromArray(body, size);
            if (ret == false) {
                LOG_ERROR("对消费到的消息进行反序列化失败！");
                return;
            }
           
            std::string filename;
            std::string fileid;
            int64_t filesize;
            std::string content;
            //3.将图片、语音、文件消息存储到文件管理服务
            switch(message.message().message_type())
            {
                //2.将文本消息存储到ES
                case MessageType::STRING :
                {
                    content = message.message().string_message().content();
                    ret = _es_message->Insert(message.sender().user_id(),
                    message.message_id(),message.timestamp(),message.chat_session_id(),
                    content);
                    if (ret == false) 
                    {
                        LOG_ERROR("对消费到的消息存储es失败！");
                        return;
                    }
                }
                break;
                case MessageType::IMAGE :
                {
                    std::string content = message.message().image_message().image_content();
                    ret = PutFile("",content,content.size(),fileid);
                    if (ret == false) 
                    {
                        LOG_ERROR("对消费到的图片消息存储文件服务失败！");
                        return;
                    }
                }
                break;
                case MessageType::FILE :
                {
                    filename = message.message().file_message().file_name();
                    filesize = message.message().file_message().file_size();
                    std::string content  = message.message().file_message().file_contents();
                    ret = PutFile(filename,content,filesize,fileid);
                    if (ret == false) 
                    {
                        LOG_ERROR("对消费到的文件消息存储文件服务失败！");
                        return;
                    }
                }
                break;
                case MessageType::SPEECH :
                {
                    std::string content = message.message().speech_message().file_contents();
                    ret = PutFile("",content,content.size(),fileid);
                    if (ret == false) 
                    {
                        LOG_ERROR("对消费到的语音消息存储文件服务失败！");
                        return;
                    }
                }
                break;
                default:
                LOG_ERROR("消息类型错误");
                return;
            }
            //4.将信息存储到Mysql
            Message msg(message.message_id(),
            boost::posix_time::from_time_t(message.timestamp()),
            message.sender().user_id(),
            message.chat_session_id(),
            message.message().message_type());
            msg.content(content);
            msg.file_id(fileid);
            msg.filename(filename);
            msg.filesize(filesize);

            ret = _mysql_message->Insert(msg);
            if(ret == false)
            {
                LOG_ERROR("对消费到的消息存储Mysql失败！");
                return;
            }

            LOG_INFO("消息存储完成 {}",message.sender().user_id());
        }

    private:
        bool PutFile(const std::string &filename,
            const std::string &body, 
            const int64_t fsize,
            std::string &file_id)
        {
            auto channel = _service_manage->Choose(_file_service_name);
            if(!channel)
            {
                LOG_ERROR("获取文件服务信道失败：{}", _file_service_name);
                return false;
            }

            xu::FileService_Stub stub(channel.get());
            xu::PutSingleFileReq req;
            xu::PutSingleFileRsp* rsp = new xu::PutSingleFileRsp();
            brpc::Controller *cntl = new brpc::Controller();

            req.set_request_id(uuid());
            req.mutable_file_data()->set_file_content(body);
            req.mutable_file_data()->set_file_name(filename);
            req.mutable_file_data()->set_file_size(fsize);
            stub.PutSingleFile(cntl,&req,rsp,nullptr);

            if(cntl->Failed() || rsp->success() == false)
            {
                LOG_ERROR("调用文件服务处理失败");
                return false;
            }

            file_id = rsp->file_info().file_id();
            return true;
        }
    private:
        ServiceManage::ptr  _service_manage;            //服务发现
        ESMessage::ptr _es_message;                     //es搜索引擎
        ManageMessage::ptr _mysql_message;              //mysql对用户数据的管理
        MQClient::ptr   _mq_client;                     //消息队列客户端

        std::string _user_service_name;                 //用户服务名称
        std::string _file_service_name;                 //文件服务名称
    };

    class MessageServer
    {
    public:
        using ptr = std::shared_ptr<MessageServer>;
        MessageServer(
            ServiceManage::ptr service_manage,
            Discovery::ptr discovery,
            Registration::ptr registration,
            std::shared_ptr<brpc::Server> server,
            std::shared_ptr<elasticlient::Client> es_client,
            std::shared_ptr<odb::mysql::database> mysql_client,
            MQClient::ptr   mq_client)
            :_discovery(discovery),
            _service_manage(service_manage),
            _registration(registration),
            _server(server),
            _es_client(es_client),
            _mysql_client(mysql_client),
            _mq_client(mq_client)
        {}
        ~MessageServer(){}
        void Start()
        {
            _server->RunUntilAskedToQuit();
        }
    private:
        ServiceManage::ptr _service_manage;                 //服务发现信道管理
        Discovery::ptr _discovery;                          //服务发现
        Registration::ptr _registration;                    //etcd注册
        std::shared_ptr<brpc::Server> _server;              //rpc注册调用
        std::shared_ptr<elasticlient::Client> _es_client;   //es搜索引擎客户端
        std::shared_ptr<odb::mysql::database> _mysql_client;//mysql客户端
        MQClient::ptr   _mq_client;                         //消息队列客户端
    };

    class MessageServerBuilder
    {
    public:
        MessageServerBuilder(){}
        ~MessageServerBuilder(){}

        //构造服务发现信道
        void SetServiceManage(const std::string &user_service_name,const std::string &file_service_name)
        {
            _user_service_name = user_service_name;
            _file_service_name = file_service_name;

            //1.创建对象
            _service_manage = std::make_shared<xu::ServiceManage>();
            //2.添加发现服务
            _service_manage->AddFollow(user_service_name);
            _service_manage->AddFollow(file_service_name);
        }
        //构造etcd发现
        void SetDiscovery(const std::string &etcd_host,const std::string &base_service)
        {
            if(!_service_manage)
            {
                LOG_ERROR("ServiceManage未初始化");
                return ;
            }
            //1.绑定回调函数
            auto put_cb = std::bind(&xu::ServiceManage::onServiceOnline, _service_manage.get(), std::placeholders::_1, std::placeholders::_2);
            auto del_cb = std::bind(&xu::ServiceManage::onServiceOffline, _service_manage.get(), std::placeholders::_1, std::placeholders::_2);
            //2. 构造服务发现对象
           _discovery = std::make_shared<xu::Discovery>(etcd_host,base_service, put_cb, del_cb);
        }
        //构造etcd注册
        void SetRegistration(const std::string &host,const std::string &service_name,
            const std::string &access_host)
        {
            _registration = std::make_shared<Registration>(host);
            _registration->AddKV(service_name,access_host);
        }

        //构造Es客户端
        void SetEsClient(const std::vector<std::string> &host_list)
        {
            _es_client = ESClientFactory::Create(host_list);
        }

        //构造Mysql客户端
        void SetMysqlClient(const std::string & user,
            const std::string &pswd,const std::string &db_name,
            const std::string &host,const std::string &cset)    
        {
            _mysql_client = MysqlClientFactory::Create(user,pswd,db_name,host,cset);
        }

        //构造消息队列
        void SetRQClient(const std::string &user, const std::string &password, const std::string &host,
        const std::string &exchange_name, const std::string &queue_name, const std::string &routing_key)
        {
            _mq_client = std::make_shared<MQClient>(user,password,host);
            _mq_client->DeclareComponents(exchange_name,queue_name,routing_key);
        }

    
        //构造rpc注册调用
        void SetRpcServer(uint16_t port, int32_t timeout, uint8_t num_threads, 
            const std::string &queue_name,
        const std::string &queue_tag)
        {
            if(!_service_manage)
            {
                LOG_ERROR("ServiceManage未初始化");
                return ;
            }

            if(!_es_client)
            {
                LOG_ERROR("EsClient未初始化");
                return ;
            }

            if(!_mysql_client)
            {
                LOG_ERROR("MysqlClient未初始化");
                return ;
            }
            
            if(!_mq_client)
            {
                LOG_ERROR("SetRQClient未初始化");
                return ;
            }

            _server = std::make_shared<brpc::Server>();
   
            MsgStorageServicelmpl *message_service = new  MsgStorageServicelmpl(_service_manage,
            _es_client, _mysql_client,_mq_client,_user_service_name,_file_service_name);

            _mq_client->Consume(queue_name, queue_tag,std::bind(&MsgStorageServicelmpl::onMessage,message_service,
            std::placeholders::_1,std::placeholders::_2));

            _server->AddService(message_service,brpc::ServiceOwnership::SERVER_OWNS_SERVICE);

            brpc::ServerOptions options;
            options.idle_timeout_sec = timeout;
            options.num_threads = num_threads;

            _server->Start(port, &options);
        }

        //构造UserServer
        MessageServer::ptr Build()
        {
            if(!_registration)
            {
                LOG_ERROR("Registration未初始化");
                return MessageServer::ptr();
            }

             if(!_discovery)
            {
                LOG_ERROR("Discovery未初始化");
                return MessageServer::ptr();
            }

            if(!_server)
            {
                LOG_ERROR("RpcServer未初始化");
                return MessageServer::ptr();
            }

            auto res = std::make_shared<MessageServer>(_service_manage,_discovery,_registration,
            _server,_es_client,_mysql_client,_mq_client);
            return res;
        }

    private:
        ServiceManage::ptr _service_manage;                 //服务发现信道管理
        Discovery::ptr _discovery;                          //服务发现
        Registration::ptr _registration;                    //etcd注册
        std::shared_ptr<elasticlient::Client> _es_client;   //es搜索引擎客户端
        std::shared_ptr<odb::mysql::database> _mysql_client;//mysql客户端
        MQClient::ptr   _mq_client;                         //消息队列客户端
        std::shared_ptr<brpc::Server> _server;              //rpc注册调用     
        std::string _user_service_name;                     //用户服务名称
        std::string _file_service_name;                     //文件服务名称
    };
}
